package com.lanhao.aot.redis;

import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;

public class RedisShardingUtil {

	/**
	 * 虚拟节点与hash值对应关系
	 */
	public static TreeMap<Long, String> virtualNodes = new TreeMap<>();
	public static Set<Long> slotHashCollections = new HashSet<>();
	public static Object lock = new Object();

	public static long hash(String key) {
		return hash(key.getBytes());
	}

	/**
	 * murmur hash算法实现
	 */
	public static long hash(byte[] key) {

		ByteBuffer buf = ByteBuffer.wrap(key);
		int seed = 0x1234ABCD;

		ByteOrder byteOrder = buf.order();
		buf.order(ByteOrder.LITTLE_ENDIAN);

		long m = 0xc6a4a7935bd1e995L;
		int r = 47;

		long h = seed ^ (buf.remaining() * m);

		long k;
		while (buf.remaining() >= 8) {
			k = buf.getLong();

			k *= m;
			k ^= k >>> r;
			k *= m;

			h ^= k;
			h *= m;
		}

		if (buf.remaining() > 0) {
			ByteBuffer finish = ByteBuffer.allocate(8).order(ByteOrder.LITTLE_ENDIAN);
			// for big-endian version, do this first:
			// finish.position(8-buf.remaining());
			finish.put(buf).rewind();
			h ^= finish.getLong();
			h *= m;
		}

		h ^= h >>> r;
		h *= m;
		h ^= h >>> r;

		buf.order(byteOrder);
		return h;
	}

	public static void initRedisGroup(int slotNo) {
		for (int i = 1; i <= slotNo; i++) {
			String slotName = Const.PREFIX_SLOT_NAME + i;
			short weight = 0; // 权重，当前可不设置，取默认值即可
			RedisShardingUtil.addSlot(slotName, RedisShardingUtil.calcVirtualNodeNum(weight));
		}
	}

	public static int calcVirtualNodeNum(short weight) {
		if (weight == 0) {
			return Const.DEFAULT_VIRTUAL_NODES;
		} else if (weight > 0) {
			return Const.DEFAULT_VIRTUAL_NODES * (weight + 1);
		}
		return Const.DEFAULT_VIRTUAL_NODES / (weight * -1 + 1);
	}

	// 注册的时候调用此方法
	public static void addSlot(String slotName, int slotNodeNum) {
		synchronized (lock) {
			for (int i = 0; i < slotNodeNum; i++) {
				String virtualNode = slotName + Const.PREFIX + i;
				long hash = RedisShardingUtil.hash(virtualNode);
				if (RedisShardingUtil.virtualNodes.containsKey(hash)) {
					throw new RuntimeException("pls check slot repetitive, virtualNode: " + virtualNode);
				}
				RedisShardingUtil.virtualNodes.put(hash, virtualNode);
				RedisShardingUtil.slotHashCollections.add(hash);
			}
		}
	}
	public static String findSlot(String key) {
		long hash = RedisShardingUtil.hash(key.getBytes());
		final String slotName;
		// 如果hash值巧合已有，直接使用此key,否则使用小于此hash值的最近一个key
		if (RedisShardingUtil.slotHashCollections.contains(hash)) {
			slotName = RedisShardingUtil.virtualNodes.get(hash);
		} else {
			Map.Entry<Long, String> longStringEntry = RedisShardingUtil.virtualNodes.lowerEntry(hash);
			if (longStringEntry == null) {
				// 碰巧是最小的hash值，取第一个
				longStringEntry = RedisShardingUtil.virtualNodes.firstEntry();
			}
			slotName = longStringEntry.getValue();
		}
		return slotName.substring(0, slotName.indexOf(Const.PREFIX));
	}
}
